package com.niit.spark.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}

/**
 * Date:2025/4/15
 * Author：Ys
 * Description:
 */
object Transform_RDD01 extends App {

  //准备环境
  val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Action_RDD")

  val sc = new SparkContext(sparkConf)
  sc.setLogLevel("ERROR") // 设置日志打印的级别 Error

  /*
   转换算子
     map:将处理的数据逐条进行映射转换，这里的转换可以是类型转换，也可以是值转换
     需求：将列表中 1 2 3 4 转换 2 4 6 8
   */
   val rdd1: RDD[Int] = sc.makeRDD(1 to 4)
   val mapRdd1: RDD[Int] = rdd1.map(_ * 2)
   mapRdd1.collect().foreach(println)


   /*
   mapPartitions:将待处理的数据以分区为单位发送道计算节点进行处理，这里的处理是指可以进行任意处理，甚至是过滤数据
   需求：找到每个分区的最大值
   */
   val rdd2: RDD[Int] = sc.makeRDD(1 to 10, 2)//【1，2，3，4，5】 【6，7，8，9，10】
   val mapRdd2: RDD[Int] = rdd2.mapPartitions(iter => {
    //iter 只是就是每个分区的数据 【1，2，3，4，5】
    List(iter.max).iterator

  })

  println(mapRdd2.collect().mkString(","))


  /*
  flatMap:将处理的数据进行拆分后进行扁平化
   */
  val rdd3 = sc.makeRDD( List( List(1,2),  List(3,4) ) ) // ==> List( 1 ,2 ,3, 4)
  val mapRdd3: RDD[Int] = rdd3.flatMap(list => list)
  mapRdd3.collect().foreach(println)
  /*
  需求：以空格拆分单词并合并
   */
  val rdd4: RDD[String] = sc.makeRDD(List("Hello Spark", "Hello Hadoop"))
  //=> List(Hello, Spark, Hello, Hadoop)

  val mapRdd4: RDD[String] = rdd4.flatMap(_.split(" "))
  mapRdd4.collect().foreach(println)


  /*
  glom:将同一个分区的数据直接转换为相同类型的数组，分区不变
   */
  val rdd5 = sc.makeRDD( List(1,2,3,4) , 2 ) //【1，2】  【3，4】
  val glomRdd1: RDD[Array[Int]] = rdd5.glom()// [   [1,2] , [3,4]          ]
  val array: Array[Array[Int]] = glomRdd1.collect()
  array.foreach( data=> println(data.mkString(",")) ) //data 是每一个一位数组
  //需求：获得两个分区最大值的和 rdd5  2 + 4
  val glomRdd2: RDD[Array[Int]] = rdd5.glom()
   val maxRdd: RDD[Int] = glomRdd2.map(array => { //array 是每一个一位数组
    array.max
  }) //[2 , 4]
  println( maxRdd.collect().sum)

  /*
  groupBy:将数据根据指定的规则进行分分，返回的RDD类型为
     分组后分区默认不变，但是数据会被打乱重新组合，我们将这样的操作称之为shuffle。
     在某些极限的情况下，数据可能被分在同一个分区中，一个组的数据在同一个分区中，但是并不是说一个分区只有一个组
   */
  val rdd6 = sc.makeRDD( List(1,2,3,4),2 ) // 奇数 一组 偶数 一组
    val groupRdd1: RDD[(Int, Iterable[Int])] = rdd6.groupBy(_ % 2)
  groupRdd1.collect().foreach(println)

  /*
   需求：按照单词的首字母进行分组
   */
  val rdd7 = sc.makeRDD( List("Hello", "Spark", "HBase,","Scala","Kafka" ) )
  // "Hello"  --->    charAt(0)        --- > H
  val groupRdd2: RDD[(Char, Iterable[String])] = rdd7.groupBy(_.charAt(0))

  groupRdd2.collect().foreach(println)


  /*
  filter:将数据根据指定的规则进行过滤，返回的RDD类型。
      数据进行过滤后不会改变分区数，但是分区内的数据可能不均衡，在生产环境下，可能会出现数据倾斜。
   需求：获得列表中的奇数
   */
  val rdd8 = sc.makeRDD( 1 to 10 )
  val filterRdd: RDD[Int] = rdd8.filter(_ % 2 != 0)
  filterRdd.collect().foreach(println)


  /*
  distinct:对RDD中的数据进行去重，返回的RDD类型不变，但是数据量会变少，分区数不变，数据不会打乱重新组合。
   */
  val rdd9 = sc.makeRDD( List(2,3,1,4,2,4,5,1,7,9) )
  val distinctRdd: RDD[Int] = rdd9.distinct(2) // 去重的时候可以设设置新RDD的分区数 2 分区数
  println("分区数：",distinctRdd.getNumPartitions)
  println(distinctRdd.collect().mkString(","))

  /*
   sortBy:对RDD中的数据进行排序,默认是升序，如果想改成降序，第二个参数设置false
   */
   val sortRDD: RDD[Int] = distinctRdd.sortBy(num => num,false)
   sortRDD.collect().foreach(println)


  //sortBy 按照Key进行排序
  val rdd10: RDD[(String, String)] = sc.makeRDD(List(("1", "a"), ("3", "c"), ("4", "d"), ("2", "b")))
  val sortRdd2: RDD[(String, String)] = rdd10.sortBy(_._1.toInt)
  sortRdd2.collect().foreach(println)

  /*
   coalesce:缩小分区，用于大数据集过滤后，提高小数据集的程序效率
     当Spark程序中，存在过多的小任务的时候，可以通过coalesce，合并分区，减少分区个数，减小任务调度成本
   */
  val rdd11 = sc.makeRDD(1 to 6, 3)
  //【1，2】 【3，4】 【5，6】
  //减少分区会发生数据倾斜，如果去避免数据倾斜？
  //将第二个参数设置为true，表示进行shuffle操作，可以解决数据倾斜问题
  val coalesceRdd: RDD[Int] = rdd11.coalesce(2,true)
  //==> 【1，2，3】  【4，5，6】
  //coalesceRdd.saveAsTextFile("output5")

  /*
   repartition:重新分区。可以增加分区，也可以缩减分区，会自动执行shuffle操作
   */
  val rdd12 = sc.makeRDD(1 to 6, 2)
  private val reRdd: RDD[Int] = rdd12.repartition(6)
  //reRdd.saveAsTextFile("output6")

  /*
   交集 并集 差集 拉链
   */
  val rdd13 = sc.makeRDD(List(1,2,3,4))
  val rdd14 = sc.makeRDD(List(3,4,5,6))
  //交集
  val inRdd: RDD[Int] = rdd13.intersection(rdd14)
  inRdd.collect().foreach(println)


  //并集
  val unionRdd: RDD[Int] = rdd13.union(rdd14)
  unionRdd.collect().foreach(println)

  //差集
  val subRdd: RDD[Int] = rdd13.subtract(rdd14)
  subRdd.collect().foreach(println)


  //拉链
  val zipRdd: RDD[(Int, Int)] = rdd13.zip(rdd14)
  println(zipRdd.collect().mkString(","))

  //partitionBy: 按照指定的分区器重新分区，Spark默认的分区器是HashPartitioner
  val rdd15 = sc.makeRDD(List(1,2,3,4),2)
  //隐式转换 ==》 (1,2,3,4) ==> (1,1) (2,1) (3,1) (4,1)
  val mapRdd5: RDD[(Int, Int)] = rdd15.map((_, 1))
  val parRdd: RDD[(Int, Int)] = mapRdd5.partitionBy(new HashPartitioner(3))
  //parRdd.saveAsTextFile("output7")

  /*
  reduceByKey:根据key进行聚合，然后进行聚合操作，返回的RDD类型为(K,V)
   */
   val rdd16: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 2), ("b", 3), ("b", 4), ("b", 5), ("a", 6)))
   val redRed: RDD[(String, Int)] = rdd16.reduceByKey((x: Int, y: Int) => {
    x + y
  })
  val redRed2: RDD[(String, Int)] =  rdd16.reduceByKey(_+_)
  redRed2.collect().foreach(println)


  /*
  groupByKey:根据key对value进行分组
   */
  val rdd17: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 2), ("b", 3), ("b", 4), ("b", 5), ("a", 6)))
  val groupRdd3: RDD[(String, Iterable[Int])] = rdd17.groupByKey()
  //===> (a,[1,2,6])  (b,[3,4,5])
  groupRdd3.collect().foreach(println)

  /*
   aggregateByKey:将数据根据不同的规则进行分区内和分区间计算，同时在使用该算子的时候要传递一个初始值，返回的RDD类型为(K,V)

   */
  val rdd18: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 2), ("b", 3), ("b", 4), ("b", 5), ("a", 6)  ) ,2 )
  //=>  【 (a,1) (a,2) (b,3) (a,0),(b,0) 】  【 (a,0),(b,0), (b,4) (b,5) (a,6) 】
  //在分区内找到每个key的最大值，再进行分区间的相同key的最大值的相加
  // 【 (a,2) (b,3)   】  【   (b,5) (a,6) 】 ==>  (a,2+6 = 8)  (b,3+5 = 8)  => (a,8)  (b,8)

  val aggRdd1: RDD[(String, Int)] = rdd18.aggregateByKey(0)(
    //第一个参数 ==> 分区内的计算规则  x和y 就表示每个分区中，相同key的值
    // x是用于计算上一次的计算结果 初始值为0  y就是a中的每一个值  0 1 ==》 x=1
    (x, y) => math.max(x, y)
    ,
    //第二个参数 ==> 分区间的计算规则 x 和 y 表示每个分区中 相同Key的最大值
    (x, y) => x + y

  )
  aggRdd1.collect().foreach(println)

  // 利用aggregateByKey，计算rdd18不同的Key和 初始值设5 ==> (a,1+2+6+5) (b,3+4+5+5)
  var aggRdd2: RDD[(String, Int)] = rdd18.aggregateByKey(5)(
    // x是用于计算上一次的计算结果 初始值为0  y就是a中的每一个值
    // a 1 2 6 5  =》 x=0 y= 1  ==> 0+1 = 1   x=1 y = 2 ==> 1+2 = 3  x=3 y = 6 ==> 3+6 = 9 x=9 y = 5 ==> 9+5 = 14
    //第一个参数 ==> 分区内的计算规则  x和y 就表示每个分区中，相同key的值
    (x, y) => x + y
    ,
    //第二个参数 ==> 分区间的计算规则 x 和 y 表示每个分区中 相同Key的值
    (x, y) => x + y
  )
  //简写方式
  aggRdd2 = rdd18.aggregateByKey(5)(_+_,_+_)
  // 再简写方式：当分区内和分区间的计算规则一样的时候，可以简写为 foldByKey的方法
  aggRdd2 = rdd18.foldByKey(5)(_+_)
  aggRdd2.collect().foreach(println)

  /*
   // 利用aggregateByKey，计算rdd18不同的Key的平均值 初始值设 (0,0) ==> 只能求出 不同Key的总和以及key出现的次数
     ==> 第一个 0 表示的是相同Key的value的和，第二个0表示的是相同Key的出现的次数
   */
  //val rdd18: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("a", 2), ("b", 3), ("b", 4), ("b", 5), ("a", 6)  ) ,2 )
  val aggrdd3: RDD[(String, (Int, Int))] = rdd18.aggregateByKey((0, 0))( // 元组根据__序号__获得里面的元素  序号从1 开始 索引 从 0 开始
    // t就是(0,0) v 就是key中的每一个值==》 a : 1 2
    // //=>  【 (a,1) (a,2) (b,3)  】                                【  (b,4) (b,5) (a,6) 】
    (t, v) => {
      (t._1 + v, t._2 + 1)
      //  1 【a:(0+1+2,0+1+1)    b:(0+3,0+1) 】              2 【 a:(0+6,0+1)   b:(0+4+5,0+1+1) 】
    },
    (t1, t2) => { // t1:第一个分区内计算结果    t2：第二个分区内的计算结果
      (t1._1 + t2._1, t1._2 + t2._2) //==》 ByKey会自动的帮我们相同key进行计算 a:(3+6,2+1)  b:(3+9,1+2)
    }
  )//==> (a,(9,3)) (b,(12,3)) ==> a 9/3  b 12/3
                                            //t (9,3)    (12,3)
   val mapRdd6: RDD[(String, Int)] = aggrdd3.mapValues(t => t._1 / t._2)  // => mapByKey
  mapRdd6.collect().foreach(println)

  /*
  sortByKey:根据key进行排序，默认是升序，第一个参数可以设置是否降序
   */
   val rdd19: RDD[(String, Int)] = sc.makeRDD(List(("a", 1), ("c", 3), ("b", 2)))
   val sortRdd3: RDD[(String, Int)] = rdd19.sortByKey(false)
  sortRdd3.collect().foreach(println)

  /*
   join:根据key进行join，返回的RDD类型为(K,(V,W))
   */
  val rdd20 = sc.makeRDD( List(  ("a",1) , ("a",2) , ("c",3) , ("b",8)  ) )
  val rdd21 = sc.makeRDD( List(  ("a",5) , ("c",6) , ("a",4) , ("g",7)  ) )

  val joinRdd: RDD[(String, (Int, Int))] = rdd20.join(rdd21)
   joinRdd.collect().foreach(println)

  val leftjoinRdd: RDD[(String, (Int, Option[Int]))] = rdd20.leftOuterJoin(rdd21)
  leftjoinRdd.collect().foreach(println)
  val rightjoinRdd: RDD[(String, (Option[Int], Int))] = rdd20.rightOuterJoin(rdd21)
  rightjoinRdd.collect().foreach(println)

  /*
    cogroup:分组后拼接  (group+connect)
   */
  val rdd22 = sc.makeRDD( List(  ("a",1) , ("b",2)  ) )
  val rdd23 = sc.makeRDD( List(  ("a",4) , ("b",5) , ("c",6) , ("c",7)  ) )
  val cgRdd: RDD[(String, (Iterable[Int], Iterable[Int]))] = rdd22.cogroup(rdd23)
  cgRdd.collect().foreach(println)


  sc.stop()

}
